package com.niodata.dt.fs.http;

import com.google.common.base.Strings;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.PasswordAuthentication;
import java.net.Proxy;
import java.net.Proxy.Type;
import java.net.URI;
import java.util.List;
import java.util.concurrent.TimeUnit;
import okhttp3.Authenticator;
import okhttp3.Credentials;
import okhttp3.HttpUrl;
import okhttp3.OkHttpClient;
import okhttp3.OkHttpClient.Builder;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.Route;
import org.apache.commons.codec.binary.Base64;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.util.Progressable;
import org.apache.log4j.Logger;
import org.jsoup.Connection;
import org.jsoup.HttpStatusException;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;

public class HttpFileSystem extends FileSystem {

  public static final String FS_HTTP_HOST = "fs.http.host";
  public static final String FS_HTTP_PORT = "fs.http.host.port";
  public static final String FS_HTTP_USER = "fs.http.user";
  public static final String FS_HTTP_PASSWORD = "fs.http.password";
  public static final String FS_HTTP_PROTOCOL = "fs.http.protocol";
  public static final String FS_HTTP_CONNECTION_TIMEOUT = "fs.http.connection.timeout";
  public static final String FS_HTTP_PROXY_URL = "fs.http.proxy.url";
  public static final String FS_HTTP_PROXY_USER = "fs.http.proxy.user";
  public static final String FS_HTTP_PROXY_PASSWORD = "fs.http.proxy.password";
  public static final String FS_HTTP_TIME_FORMAT = "fs.http.time.format";
  private static final Logger logger = Logger.getLogger(HttpFileSystem.class);
  private String base64login = null;


  private Path workingDir = new Path("/");

  private URI uri;
  private String protocol;
  private int port;

  @Override
  public void initialize(URI uri, Configuration conf) throws IOException {
    super.initialize(uri, conf);
    this.uri = uri;
    // get host information from uri (overrides info in conf)
    String host = uri.getHost();
    host = (host == null) ? conf.get(FS_HTTP_HOST, null) : host;
    if (host == null) {
      throw new IOException("Invalid host specified");
    }
    conf.set(FS_HTTP_HOST, host);
    this.protocol = uri.getScheme();
    this.port = uri.getPort();
    if (this.port == -1) {
      if ("http".equalsIgnoreCase(protocol)) {
        this.port = 80;
      } else {
        this.port = 443;
      }
    }
    setConf(conf);
    this.uri = uri;
    String user = getConf().get(FS_HTTP_USER, null);
    if (!Strings.isNullOrEmpty(user)) {
      String login = user + ":" + getConf().get(FS_HTTP_PASSWORD);
      base64login = new String(Base64.encodeBase64(login.getBytes()));
    }
  }

  /**
   * Return the protocol scheme for the FileSystem.
   * <p/>
   *
   * @return <code>ftp</code>
   */
  @Override
  public String getScheme() {
    return this.protocol;
  }

  @Override
  public URI getUri() {
    return uri;
  }

  @Override
  public int getDefaultPort() {
    return port;
  }


  @Override
  public FSDataInputStream open(Path path, int bufferSize) throws IOException {
    FileStatus status = getFileStatus(path);
    if (status.isDirectory()) {
      throw new IOException("path is directory");
    }
    Request.Builder builder = new Request.Builder();
    String url = uri.toString() + path;
    builder.url(HttpUrl.parse(url));
    builder.method("GET", null);
    if (base64login != null) {
      builder.header("authorization", "Basic " + base64login);
    }
    Request request = builder.build();
    OkHttpClient client = buildClient();
    Response response = client.newCall(request).execute();
    HttpFsInputStream inputStream = new HttpFsInputStream(response.body().byteStream());
    return new FSDataInputStream(inputStream);
  }

  @Override
  public FSDataOutputStream create(Path path, FsPermission fsPermission, boolean b, int i, short i1,
        long l, Progressable progressable) throws IOException {
    throw new IOException("operation not supported");
  }

  @Override
  public FSDataOutputStream append(Path path, int i, Progressable progressable) throws IOException {
    throw new IOException("operation not supported");
  }

  @Override
  public boolean rename(Path path, Path path1) throws IOException {
    throw new IOException("operation not supported");
  }

  @Override
  public boolean delete(Path path, boolean b) throws IOException {
    throw new IOException("operation not supported");
  }

  @Override
  public FileStatus[] listStatus(Path path) throws IOException {
    String url = uri.toString() + path.toString();
    List<FileStatus> statusList = null;
    try {
      Document document = getDocument(url);
      //document to status
      statusList = HttpFileUtil.parseFile(document, path.toString(), getConf());
    } catch (HttpStatusException exception) {
      if (exception.getStatusCode() == 404) {
        throw new FileNotFoundException(path.toString());
      }
      if (exception.getStatusCode() == 401) {
        throw new RuntimeException("http auth failed,status=401:" + exception.getMessage());
      }
    } catch (Exception ex) {
      ex.printStackTrace();
      throw new IOException(ex);
    }
    return statusList.toArray(new FileStatus[statusList.size()]);
  }

  @Override
  public Path getWorkingDirectory() {
    return workingDir;
  }

  @Override
  public void setWorkingDirectory(Path path) {
    this.workingDir = path;
  }

  @Override
  public boolean mkdirs(Path path, FsPermission fsPermission) throws IOException {
    throw new IOException("operation not supported");
  }

  @Override
  public FileStatus getFileStatus(Path path) throws IOException {
    FileStatus[] statuses = this.listStatus(path.getParent());
    for (FileStatus status : statuses) {
      if (status.getPath().equals(path)) {
        return status;
      }
    }
    throw new FileNotFoundException(path.toString());
  }

  private boolean isUseProxy() {
    String ps = super.getConf().get(FS_HTTP_PROXY_URL, "");
    return !Strings.isNullOrEmpty(ps);
  }

  private String getProxyPassword() {
    String pwd = super.getConf().get(FS_HTTP_PROXY_PASSWORD, "");
    return pwd;

  }

  private String getProxyUser() {
    String user = super.getConf().get(FS_HTTP_PROXY_USER, "");
    return user;
  }

  private String getFsHttpProxyUrl() {
    String ps = super.getConf().get(FS_HTTP_PROXY_URL, "");
    return ps;
  }

  private Document getDocument(String path) throws Exception {
    Connection connection = null;
    if (base64login != null) {
      connection = Jsoup.connect(path)
            .timeout(getConf().getInt(FS_HTTP_CONNECTION_TIMEOUT, 30 * 1000))
            .header("authorization", "Basic " + base64login);
    } else {
      connection = Jsoup.connect(path)
            .timeout(getConf().getInt(FS_HTTP_CONNECTION_TIMEOUT, 30 * 1000));
    }
    connection.header("user-agent",
          "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_3) "
                + "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.114 Safari/537.36");
    if (isUseProxy()) {
      String ps = getFsHttpProxyUrl();
      Type type = ps.startsWith("http") ? Type.HTTP : Type.SOCKS;
      //http://127.0.0.1:80
      String[] ss = ps.split("//")[1].split(":");
      InetSocketAddress address = new InetSocketAddress(ss[0],
            ss.length > 1 ? Integer.parseInt(ss[1]) : 80);
      Proxy proxy = new Proxy(type, address);
      connection.proxy(proxy);
      if (!Strings.isNullOrEmpty(getProxyUser())) {
        java.net.Authenticator.setDefault(
              new java.net.Authenticator() {
                public PasswordAuthentication getPasswordAuthentication() {
                  return new PasswordAuthentication(
                        getProxyUser(), getProxyPassword().toCharArray());
                }
              }
        );
      }
    }

    for (int i = 0; i < 3; i++) {
      try {
        Document document = connection.timeout(120 * 1000).get();
        return document;
      } catch (Exception e) {
        if (i == 2) {
          throw e;
        } else {
          logger.error(e);
          Thread.sleep(5 * 1000);
        }
      }
    }
    return null;
  }

  private OkHttpClient buildClient() throws IOException {
    OkHttpClient.Builder clientBuilder = new Builder();
    clientBuilder.connectTimeout(10, TimeUnit.SECONDS)
          .readTimeout(20, TimeUnit.SECONDS)
          .writeTimeout(20, TimeUnit.SECONDS);
    String ps = super.getConf().get(FS_HTTP_PROXY_URL, "");
    if (!Strings.isNullOrEmpty(ps)) {
      Type type = ps.startsWith("http") ? Type.HTTP : Type.SOCKS;
      //http://127.0.0.1:80
      String[] ss = ps.split("//")[1].split(":");
      InetSocketAddress address = new InetSocketAddress(ss[0],
            ss.length > 1 ? Integer.parseInt(ss[1]) : 80);
      Proxy proxy = new Proxy(type, address);
      clientBuilder.proxy(proxy);
      String user = getConf().get(FS_HTTP_PROXY_USER, "");
      String password = getConf().get(FS_HTTP_PROXY_PASSWORD, "");
      if (!Strings.isNullOrEmpty(user) && !Strings.isNullOrEmpty(password)) {
        okhttp3.Authenticator proxyAuthenticator = new Authenticator() {
          @Override
          public Request authenticate(Route route, Response response) throws IOException {
            String credential = Credentials
                  .basic(user, password);
            return response.request().newBuilder()
                  .header("Proxy-Authorization", credential)
                  .build();
          }
        };
        clientBuilder.authenticator(proxyAuthenticator);
      }
    }
    return clientBuilder.build();
  }
}
